/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.server;

import static org.sss.common.Utils.closeDataReceivers;

import java.lang.reflect.Method;
import java.util.List;
import java.util.UUID;

import org.slf4j.Logger;
import org.sss.common.Bucket;
import org.sss.common.CommonStatus;
import org.sss.common.DataReceiver;
import org.sss.common.Inlet;
import org.sss.common.TaskSliceSubmitter;
import org.sss.common.io.DataReader;
import org.sss.common.io.KeyFormatter;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.Decoder;
import org.sss.mapreduce.Mapper;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.StorageNode;
import org.sss.util.CallbackInterface;

public class MapTask extends MapTaskBase {
  private static Logger logger = SssServer.getLogger(MapTask.class);

  public MapTask(TaskSet taskSet,
                 UUID jobID,
                 SssServerResource sssResource,
                 Configuration conf,
                 StorageNode storageNode,
                 Inlet inlet,
                 Class<? extends Mapper> mapperClass,
                 List<org.sss.common.Outlet> outletList,
                 CallbackInterface<CommonStatus> doneCallback) throws SssException {
    super(taskSet, jobID, sssResource, conf, storageNode, inlet,
        mapperClass, outletList, doneCallback);
  }

  @Override
  protected synchronized void submitReadTaskSlices() throws SssException {
    for (int i = 0; i < getInputGID().getNSlot(); i++) {
      DataReader r = getIOManager().createDataReader(
          getConfiguration(), getStorageNode(), getInputGID(), i,
          new TaskSliceSubmitter<List<byte[]>>(getWKTP(), this));
      submitReadRequest(r);
    }
  }

  /**
   * Execute user map function.
   *
   * NOTE: This run in a worker thread.
   */
  @Override
  public void handle(List<byte[]> kvs) {
    try {
      Method mapMethod = getProcInfo().getMethod();
      Mapper mapper = getProc();
      logger.debug("run {} tuples.", kvs.size()/2);
      SssServerConfiguration sssConf = getSssServerConfiguration();

      List<DataReceiver<Bucket>> cache = getOutlets().getTLPutters(sssConf.map_output_cache_nbytes);
      Object[] args = createArguments(createContext(), 2, cache);

      Decoder keyDecoder = getKeyDecoder();
      Decoder valDecoder = getValueDecoder();
      KeyFormatter kf = KeyFormatter.get(getInputGID().getKeyFormat());
      for (int i = 0; i < kvs.size(); i += 2) {
        checkCanceled();
        if (sssConf.ioConf.stream_protocol) {
          args[1] = keyDecoder.decode(kvs.get(i));
        }
        else {
          args[1] = keyDecoder.decode(kf.getUserKey(kvs.get(i)));
        }
        args[2] = valDecoder.decode(kvs.get(i + 1));

        mapMethod.invoke(mapper, args);
      }
      closeDataReceivers(cache);
    }
    catch (Throwable e) {
      handleExceptionInHandle(e, "Error in Map Task, Cancels tasks");
    }
    finally {
      logger.debug("completing a MapTaskSlice.");
    }
  }
}
